package gossip.v2.protocol;

import gossip.v2.GossipCore;
import gossip.v2.message.*;
import gossip.v2.cluster.GossipNode;
import lombok.extern.slf4j.Slf4j;
import org.ricks.ioc.utils.Logger;
import org.ricks.net.StateMachineEnum;
import org.ricks.net.extension.processor.AbstractMessageProcessor;
import org.ricks.net.transport.AioSession;

import java.io.IOException;
import java.util.*;

// 消息处理器
@Slf4j
public class GossipMessageProcessor extends AbstractMessageProcessor<GossipMessage> {
    private final GossipCore core;

    public GossipMessageProcessor(GossipCore core) {
        this.core = core;
    }

    @Override
    public void process0(AioSession session, GossipMessage message) {
        message.resolveSender(core.getClusterManager());

        switch (message.getType()) {
            case ACK -> processAckMessage((AckMessage) message);
            case HEARTBEAT -> processHeartbeat(session, (HeartbeatMessage) message);
            case SYNC -> processSyncMessage(session, (SyncMessage) message);
            case USER -> processUserMessage((UserMessage) message);
            case HANDSHAKE -> processHandshake(session, (HandshakeMessage) message);
            case INDIRECT_CHECK -> processIndirectCheck((IndirectCheckRequest) message);
            default -> Logger.info("{} - 收到未知消息 - {}", session, message.getType());
        }
    }

    private void processHeartbeat(AioSession session, HeartbeatMessage msg) {
        core.getTransport().sendHeartbeat(session, HeartbeatMessage.HeartbeatType.RESPONSE);
        if (msg.getSender() != null) {
            Logger.debug("{} - 收到心跳 {}发送心跳", core.getLocalNode().getNodeId(), msg.getSenderId());
            core.getClusterManager().onHeartbeat(msg.getSender(), msg.getTimestamp());
        } else {
            Logger.error("{} - 收到了未知人发来得心跳", null, core.getLocalNode().getNodeId());
        }
    }

    private void processSyncMessage(AioSession session, SyncMessage msg) {
        if (msg.isFullSync()) {
            core.getStateManager().applyFullState(
                    msg.getVersions(),
                    msg.getStateValues()
            );
        } else {
            Map<String, byte[]> reverseDelta = core.getStateManager().mergeState(
                    msg.getVersions(),
                    msg.getStateValues()
            );

            if (!reverseDelta.isEmpty()) {
                core.getTransport().send(msg.getSender(), new AckMessage(core.getLocalNode(), reverseDelta));
            }
        }
    }

    private void processAckMessage(AckMessage msg) {
        if (msg.isIndirectResponse()) {
            core.getClusterManager().handleIndirectResponse(
                    msg.getTargetNodeId(),
                    msg.getAlive()
            );
        } else if (msg.getStateValues() != null) {
            core.getStateManager().mergeState(
                    Collections.emptyMap(),
                    msg.getStateValues()
            );
        }
    }

    private void processUserMessage(UserMessage msg) {
        core.getStateManager().update(msg.getKey(), msg.getValue());
    }

    private void processHandshake(AioSession session, HandshakeMessage msg) {
        // 1. 验证发送方信息
        if (msg.getSender() == null || msg.getSenderId() == null) {
            Logger.info("握手失败 主动关闭 {}", session);
            session.close(true);
            return;
        }

        // 2. 创建/更新节点对象
        GossipNode remoteNode = core.getClusterManager().getNodeById(msg.getSenderId());
        if (remoteNode == null) {
            remoteNode = new GossipNode(msg.getSenderId(), msg.getIp(), msg.getPort());
        } else {
            remoteNode.setIp(msg.getIp());
            remoteNode.setPort(msg.getPort());
        }

        // 检查地址冲突
        if (!msg.getIp().equals(remoteNode.getIp()) || msg.getPort() != remoteNode.getPort()) {
            Logger.warn("Address conflict detected for node {}: expected {}:{}, actual {}:{}",
                    remoteNode.getNodeId(),
                    remoteNode.getIp(), remoteNode.getPort(),
                    msg.getIp(), msg.getPort());

            // 更新为最新地址
            remoteNode.setIp(msg.getIp());
            remoteNode.setPort(msg.getPort());
        }

        // 注册会话（统一处理）
        core.getTransport().registerIncomingSession(remoteNode, session);

        // 4. 处理已知节点 (仅当是REQUEST时才处理节点列表)
        if (msg.getHandshakeType() == HandshakeMessage.HandshakeType.REQUEST) {
            for (GossipNode node : msg.getKnownNodes().values()) {
                if (node.getNodeId().equals(core.getLocalNode().getNodeId())) continue;

                if (!core.getClusterManager().isKnownNode(node.getNodeId())) {
                    core.getClusterManager().addDiscoveredNode(node);
                } else {
                    GossipNode existing = core.getClusterManager().getNodeById(node.getNodeId());
                    if (existing != null) {
                        existing.setIp(node.getIp());
                        existing.setPort(node.getPort());
                    }
                }
            }
        }

        // 5. 如果是请求才发送响应
        if (msg.getHandshakeType() == HandshakeMessage.HandshakeType.REQUEST) {
            Map<String, GossipNode> localNodes = core.getClusterManager().getAllKnownNodesWithAddress();
            // 发送响应类型握手消息
            core.getTransport().send(remoteNode,
                    new HandshakeMessage(
                            core.getLocalNode(),
                            localNodes,
                            HandshakeMessage.HandshakeType.RESPONSE
                    )
            );
        }

        // 添加连接健康检查
        if (!core.getTransport().isConnectionHealthy(remoteNode.getNodeId())) {
            Logger.warn("Connection to {} is unhealthy, scheduling reconnect", remoteNode.getNodeId());
            core.getTransport().scheduleReconnect(remoteNode);
        }

//        Logger.info("{} - Handshake completed with {}",core.getLocalNode().getNodeId(), remoteNode.getNodeId());
    }


    private void processIndirectCheck(IndirectCheckRequest msg) {
        if (core.isStopping()) return;

        boolean isAlive = false;
        if (msg.getTargetNodeId() != null) {
            isAlive = core.getClusterManager().getLiveNodes()
                    .containsKey(msg.getTargetNodeId());
        }

        core.getTransport().send(msg.getSender(),
                new AckMessage(core.getLocalNode(), core.getClusterManager().getNodeById(msg.getTargetNodeId()), isAlive));
    }

    @Override
    public void stateEvent0(AioSession session, StateMachineEnum state, Throwable throwable) {
        switch (state) {
            case NEW_SESSION -> {
                // 触发握手流程
                GossipNode localNode = core.getLocalNode();
                Map<String, GossipNode> knownNodes = core.getClusterManager().getAllKnownNodesWithAddress();
                HandshakeMessage handshake = new HandshakeMessage(
                        localNode,
                        knownNodes,
                        HandshakeMessage.HandshakeType.REQUEST
                );
                core.getTransport().sendDirect(session, handshake);
            }
            case SESSION_CLOSED -> {
                GossipNode gossipNode = core.getClusterManager().resolveNode(session);
                Logger.info("{} - 触发 SESSION_CLOSED 事件 - {}", gossipNode.getNodeId(), session.hashCode());
                if(!core.isStopping()) core.getTransport().onSessionClosed(session);
                core.getTransport().onReconnectFailure(gossipNode.getNodeId());
            }
        }
    }
}
